package com.redis.delayQueue;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import redis.clients.jedis.Jedis;

import java.lang.reflect.Type;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;


/**
 * Created with IntelliJ IDEA.
 * Description:
 *
 * @author lzy
 * @date 2020-09-22 下午5:10
 */
public class RedisDelayingQueue<T> {
    // fastjson序列化对象中存在generic类型时，需要使用TypeReference
    private final Type TaskType = new TypeReference<TaskItem<T>>() {
    }.getType();
    private final Jedis jedis;
    private final String queueKey;

    public RedisDelayingQueue(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }

    public void delay(T msg) {
        TaskItem<T> task = new TaskItem<>();
        // 分配唯一的uuid
        task.id = UUID.randomUUID().toString();
        task.msg = msg;
        // fastjson序列化
        String json = JSON.toJSONString(task);
        // 塞入队列后5s后再试
        jedis.zadd(queueKey, System.currentTimeMillis() + 5000, json);
    }

    public static void main(String[] args) {
        Jedis jedis = new Jedis("10.2.32.104", 6379);
        RedisDelayingQueue<String> queue = new RedisDelayingQueue<>(jedis, "q-demo");

        Thread producer = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                queue.delay("codehole" + i);
            }
        });

        Thread consumer = new Thread(() -> queue.loop());

        producer.start();
        consumer.start();

        try {
            producer.join();
            TimeUnit.SECONDS.sleep(6);
            consumer.interrupt();
            consumer.join();
        } catch (InterruptedException e) {
        }
    }

    public void loop() {
        while (!Thread.interrupted()) {
            // 只取出一条
            final Set<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
            if (values.isEmpty()) {
                try {
                    TimeUnit.MILLISECONDS.sleep(500L);
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }

            String s = values.iterator().next();
            // 抢到了任务
            if (jedis.zrem(queueKey, s) > 0) {
                // json反序列化
                TaskItem<T> task = JSON.parseObject(s, TaskType);
                this.handleMsg(task.msg);
            }
        }
    }

    public void handleMsg(T msg) {
        System.out.println(msg);
    }

    /**
     * 需要提交的任务对象
     *
     * @param <T>
     */
    static class TaskItem<T> {
        private String id;
        private T msg;

        public String getId() {
            return id;
        }

        public void setId(String id) {
            this.id = id;
        }

        public T getMsg() {
            return msg;
        }

        public void setMsg(T msg) {
            this.msg = msg;
        }
    }
}
